package com.atguigu.flink.chapter05.Transform;

import com.atguigu.flink.chapter05.Source.WaterSensor;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.functions.RichMapFunction;
import org.apache.flink.api.common.functions.RuntimeContext;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

/**
 * TODO
 *
 * @author cjp
 * @version 1.0
 * @date 2021/1/20 11:44
 */
public class Flink02_Map_Rich {
    public static void main(String[] args) throws Exception {
        // 获取执行环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(2);

        DataStreamSource<String> sensorDS = env
                .socketTextStream("localhost", 9999);
//        DataStreamSource<String> sensorDS = env.readTextFile("input/sensor.txt");
/*        DataStreamSource<String> sensorDS = env.fromElements(
                "sensor_1,11,1",
                "sensor_1,12,2",
                "sensor_1,13,3");*/

        SingleOutputStreamOperator<WaterSensor> resultDS = sensorDS.map(new MyMapRichFunction());

        resultDS.print();


        env.execute();
    }

    /**
     * 富函数：
     * 1. 生命周期管理方法:
     *          open： 初始化
     *          close： 关闭
     *          每个 并行实例 都会执行一次 open、close
     *     注意： 读文件时，每个并行实例，会调用2次close
     * 2. 运行时上下文 : 直接调用 getRuntimeContext() 获取上下文对象，这个对象可以获取一些其他的环境信息
     */
    public static class MyMapRichFunction extends RichMapFunction<String, WaterSensor> {

        @Override
        public void open(Configuration parameters) throws Exception {
            System.out.println("open.....");
        }

        @Override
        public WaterSensor map(String value) throws Exception {
            String[] split = value.split(",");
//            return new WaterSensor(split[0], Long.valueOf(split[1]), Integer.valueOf(split[2]));
            return new WaterSensor(getRuntimeContext().getTaskName() + " ---- " + getRuntimeContext().getTaskNameWithSubtasks(), Long.valueOf(split[1]), Integer.valueOf(split[2]));
        }
    }
}
/**
 * sensor_1,1,1
 */